package cn.cmft.jc.source;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.Objects;

public class MonitorMySQLSource {

    private final String host;
    private final int port;
    private final String username;
    private final String password;
    private final String databaseList;
    private final String tableList;

    private DataStreamSource<String> streamSource;

    public MonitorMySQLSource(ParameterTool tool){
        this.host = tool.get("monitor.datasource.hostname");
        this.port = tool.getInt("monitor.datasource.port");
        this.username = tool.get("monitor.datasource.username");
        this.password = tool.get("monitor.datasource.password");
        this.databaseList = tool.get("monitor.datasource.databaseList");
        this.tableList = tool.get("monitor.datasource.tableList");
    }
    public DataStreamSource<String> getMySQLSource(StreamExecutionEnvironment env){
        if (Objects.isNull(streamSource)){
            MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                    .hostname(host)
                    .port(port)
                    .databaseList(databaseList) // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                    .tableList(tableList) // 设置捕获的表
                    .username(username)
                    .password(password)
                    .includeSchemaChanges(true)
                    .startupOptions(StartupOptions.latest()) // 最新的
                    .heartbeatInterval(Duration.ofMillis(3000))
                    .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                    .build();
            String sourceName = MonitorMySQLSource.class.toString();
            streamSource = env.fromSource(mySqlSource,
                    WatermarkStrategy.noWatermarks(),
                    sourceName);
            return streamSource;
        }
        return streamSource;
    }
}
